package com.zh.freechat.infrastructure.mq;


import com.rabbitmq.client.Channel;
import com.zh.freechat.assemble.ChatMsgWrapper;
import com.zh.freechat.domain.chat.model.ChatMessage;
import com.zh.freechat.domain.chat.model.MsgStatus;
import com.zh.freechat.domain.chat.model.MsgStatusChange;
import com.zh.freechat.domain.message.entity.ChatMsgPO;
import com.zh.freechat.domain.message.repository.ChatMsgPoRepository;
import com.zh.freechat.domain.message.repository.ChatMsgSendRecordRepository;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.io.IOException;
import java.util.Map;

import static com.zh.freechat.common.CustomConst.MsgDataSaveQueue;
import static com.zh.freechat.common.CustomConst.MsgOfflineQueue;
import static com.zh.freechat.common.CustomConst.MsgStatusQueue;
import static com.zh.webcommon.CommonConst.GSON;


/**
 * @author Zhou
 * @date 18:10 2024/3/30
 */
@Slf4j
@AllArgsConstructor
@Component
public class IMessageMQConsumer {
    private final ChatMsgSendRecordRepository chatMsgSendRecordRepository;
    private final ChatMsgPoRepository chatMsgPoRepository;
    private final TransactionTemplate transactionTemplate;

    /**
     * 监听"聊天消息持久化"的MQ
     */
    @RabbitListener(queues = MsgDataSaveQueue, ackMode="AUTO", concurrency = "8")
    @RabbitHandler
    public void receiveMsgSaveMQ(@Payload ChatMessage message, @Headers Map<String, Object> header, Channel channel) {
        long deliveryTag = (long) header.get(AmqpHeaders.DELIVERY_TAG);

        log.info("【已接收到来自 {}的MQ消息】message[id={}]， deliveryTag={}", MsgDataSaveQueue, message.getId(), deliveryTag);

        // 批量转换
        ChatMsgPO messagePO = ChatMsgWrapper.convertEntity(message);
        messagePO.resetStatus();

        try {
            //开启事务
            transactionTemplate.execute(status -> {
                try {
                    // 批量插入数据库（事务内执行）
                    chatMsgPoRepository.insert(messagePO);
                    return true; // 执行成功
                } catch (Exception e) {
                    status.setRollbackOnly(); // 标记事务回滚
                    throw new RuntimeException("批量插入数据库失败，触发事务回滚", e); // 抛出异常，外层捕获
                }
            });

            // 批量确认消息, true表示批量确认
//            channel.basicAck(deliveryTag, true);
        } catch (Exception ex) {
            log.error("消费来自'聊天消息持久化{}'的MQ消息时，发生异常. deliveryTag = {}  error：{} \t ", MsgDataSaveQueue, deliveryTag, ex.getMessage());

//            try {
//                channel.basicNack(deliveryTag, true, false);
//            } catch (IOException ignored) {
//            }
        }
    }

    /**
     * 监听"聊天消息离线"的MQ
     */
    @RabbitListener(queues = MsgOfflineQueue, ackMode="AUTO", concurrency = "8")
    @RabbitHandler
    public void receiveMsgOfflineMQ(@Payload ChatMessage message, @Headers Map<String, Object> header, Channel channel) {
        long deliveryTag = (long) header.get(AmqpHeaders.DELIVERY_TAG);

        log.info("【已接收到来自 {}的MQ消息】message[id={}]， deliveryTag={}", MsgOfflineQueue, message.getId(), deliveryTag);

        try {

            //开启事务
            transactionTemplate.execute(status -> {
                try {
                    // 批量插入数据库（事务内执行）
                    chatMsgSendRecordRepository.insertOfflineRecord(message);

                    return true; // 执行成功
                } catch (Exception e) {
                    status.setRollbackOnly(); // 标记事务回滚
                    throw new RuntimeException("批量插入数据库失败，触发事务回滚", e); // 抛出异常，外层捕获
                }
            });

            // 批量确认消息, true表示批量确认
//            channel.basicAck(deliveryTag, true);
        } catch (Exception ex) {
            log.error("消费来自'聊天消息离线{}'的MQ消息时，发生异常. deliveryTag = {}  error：{} \t ", MsgDataSaveQueue, deliveryTag, ex.getMessage());

//            try {
//                channel.basicNack(deliveryTag, true, false);
//            } catch (IOException ignored) {
//            }
        }
    }

    /**
     * 监听"聊天消息状态更新"的MQ
     */
    @RabbitListener(queues = MsgStatusQueue, ackMode="AUTO", concurrency = "8")
    @RabbitHandler
    public void receiveMsgStatusMQ(@Payload MsgStatusChange statusChange, @Headers Map<String, Object> header, Channel channel) {
        var json = GSON.toJson(statusChange);
        long deliveryTag = (long) header.get(AmqpHeaders.DELIVERY_TAG);

        log.info("【已接收到来自 {}的MQ消息】 deliveryTag={}, \t statusChange = {}", MsgStatusQueue, deliveryTag, json);

        try {
            //开启事务
            transactionTemplate.execute(status -> {
                try {
                    doConsumeMsgStatusChange(statusChange);
                    return true; // 执行成功
                } catch (Exception e) {
                    status.setRollbackOnly(); // 标记事务回滚
                    throw new RuntimeException("批量更新数据库失败，触发事务回滚", e); // 抛出异常，外层捕获
                }
            });

            // 批量确认消息, true表示批量确认
//            channel.basicAck(deliveryTag, true);
        } catch (Exception ex) {
            log.error("消费来自'聊天消息状态更新{}'的MQ消息时，发生异常. deliveryTag = {}  error：{} \t ", MsgDataSaveQueue, deliveryTag, ex.getMessage());

//            try {
//                channel.basicNack(deliveryTag, true, false);
//            } catch (IOException ignored) {
//            }
        }
    }

    private void doConsumeMsgStatusChange(MsgStatusChange statusChange) {
        MsgStatus msgStatus = MsgStatus.of(statusChange.getMsgStatus());
        long msgId = statusChange.getMsgId();

        switch (msgStatus) {
            case SENT -> {
                log.debug("markSentRecord ... msgId = {} ", msgId);

                var receiver = statusChange.getExtra().getOrDefault("receiver", "");
                var isSuccess = "true".equalsIgnoreCase(statusChange.getExtra().getOrDefault("isSuccess", "true"));

                //保存msg的发送记录，只记录msgID、实际的接收人
                chatMsgSendRecordRepository.insertSentRecord(msgId, receiver, isSuccess);
                //离线消息表中去删除这个消息
                chatMsgSendRecordRepository.deleteOfflineRecord(msgId, receiver);
            }
            case READ -> {
                log.debug("addReadSign ... msgId = {} ", msgId);

                //此userId已读了message
                var userId = statusChange.getExtra().getOrDefault("userId", "");
                chatMsgPoRepository.addReadSign(msgId, userId);
            }
        }

        log.debug("updateStatus ... msgId = {}, msgStatus = {} ", msgId, msgStatus);
        chatMsgPoRepository.updateStatus(msgId, msgStatus.getValue());

    }
}


